/*
    信道相关类的实现：信道类、信道管理类
    1. 信道类：信道的描述，信道是在链接的基础上进一步细分，让所有的信道共用同一个链接进行通信，一个消费者对应一个信道
        属性：信道唯一标识、消费者、消费者管理句柄、protobuf协议处理器句柄、TcpConnection句柄、虚拟机
        方法：声明交换机、声明消息队列、绑定、解绑、发布、确认、订阅、取消订阅
    注：消费者：当有消费者请求的时候，这个属性才有效
*/

#pragma once
#include "muduo/net/TcpConnection.h"
#include "proto/codec.h"
#include "../../src/common/protocol.pb.h"
#include "../../src/common/threadpool.hpp"
#include "route.hpp"
#include "consumer.hpp"
#include "virtual_host.hpp"

namespace mq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;

    using OpenChannelRequestPtr = std::shared_ptr<OpenChannelRequest>;
    using CloseChannelRequestPtr = std::shared_ptr<CloseChannelRequest>;
    using DeclareExchangerRequestPtr = std::shared_ptr<DeclareExchangerRequest>;
    using RemoveExchangerRequestPtr = std::shared_ptr<RemoveExchangerRequest>;
    using DeclareQueueRequestPtr = std::shared_ptr<DeclareQueueRequest>;
    using RemoveQueueRequestPtr = std::shared_ptr<RemoveQueueRequest>;
    using BindRequestPtr = std::shared_ptr<BindRequest>;
    using UnbindRequestPtr = std::shared_ptr<UnbindRequest>;
    using PublishRequestPtr = std::shared_ptr<PublishRequest>;
    using AckRequestPtr = std::shared_ptr<AckRequest>;
    using ConsumeRequestPtr = std::shared_ptr<ConsumeRequest>;
    using CancelRequestPtr = std::shared_ptr<CancelRequest>;

    class Channel
    {
    public:
        using ChannelPtr = std::shared_ptr<Channel>;

    public:
        Channel(const std::string &id, const ConsumerManager::ConsumerManagerPtr &consumer_manager_ptr, 
            const ProtobufCodecPtr &codec_ptr, const muduo::net::TcpConnectionPtr& conn_ptr, 
            const VirtualHost::VirtualHostPtr &virtual_host_ptr, const ThreadPool::ThreadPoolPtr &thread_pool_ptr)
            : _id(id), _consumer_manager_ptr(consumer_manager_ptr), _codec_ptr(codec_ptr), 
            _conn_ptr(conn_ptr), _virtual_host_ptr(virtual_host_ptr), _thread_pool_ptr(thread_pool_ptr)
        {
        }

        ~Channel()
        {
            // 管道关闭的时候，应该删除掉它拥有的消费者
            if(_consumer_ptr != nullptr)
                _consumer_manager_ptr->removeConsumer(_consumer_ptr->_queue_name, _consumer_ptr->_tag);
        }

        // 声明交换机
        void declareExchanger(const DeclareExchangerRequestPtr &req)
        {
            bool ret = _virtual_host_ptr->declareExchanger(req->exchanger_name(), req->exchanger_type(), req->durable(), req->auto_delete(), req->args());
            responseHandler(ret, req->request_id(), req->channel_id());
        }

        // 交换机的删除
        void removeExchanger(const RemoveExchangerRequestPtr &req)
        {
            _virtual_host_ptr->removeExchanger(req->exchanger_name());
            responseHandler(true, req->request_id(), req->channel_id());
        }

        // 消息队列的声明
        void declareMsgQueue(const DeclareQueueRequestPtr &req)
        {
            bool ret = _virtual_host_ptr->declareMsgQueue(req->queue_name(), req->durable(), req->exclusive(), req->auto_delete(), req->args());
            responseHandler(ret, req->request_id(), req->channel_id());
        }

        // 删除消息队列
        void removeMsgQueue(const RemoveQueueRequestPtr &req)
        {
            // 1. 删除消息队列相关信息
            _virtual_host_ptr->removeMsgQueue(req->queue_name());

            // 2. 删除消费者队列
            _consumer_manager_ptr->destroyConsumerQueue(req->queue_name());

            responseHandler(true, req->request_id(), req->channel_id());
        }

        // 绑定
        void bind(const BindRequestPtr &req)
        {
            bool ret = _virtual_host_ptr->bind(req->exchanger_name(), req->queue_name(), req->binding_key());
            responseHandler(ret, req->request_id(), req->channel_id());
        }

        // 解绑
        void unbind(const UnbindRequestPtr &req)
        {
            _virtual_host_ptr->unbind(req->exchanger_name(), req->queue_name());
            responseHandler(true, req->request_id(), req->channel_id());
        }

        // 消息的发布
        void publish(const PublishRequestPtr &req)
        {
            // 1. 判断交换机是否存在
            Exchanger::ExchangerPtr exchanger_ptr = _virtual_host_ptr->getExchanger(req->exchanger_name());
            if(exchanger_ptr == nullptr)
                responseHandler(false, req->request_id(), req->channel_id());
            
            // 2. 获取交换机的所有绑定信息
            BindingManager::MsgQueueBindingMap msg_queue_bind_map = _virtual_host_ptr->exchangerBindings(req->exchanger_name());

            // 3. 路由
            DBG_LOG("publish message to exchanger: %s\n", req->exchanger_name().c_str());
            BasicProperties* properties_ptr = req->has_properties() ? req->mutable_properties() : nullptr; 
            std::string routing_key = properties_ptr ? properties_ptr->routing_key() : "";
            // DBG_LOG("MsgQueueBindingMap size: %ld\n", msg_queue_bind_map.size());
            for(auto& binding : msg_queue_bind_map) // 循环判断消息可以发布到交换机绑定的哪个队列中
            {
                if(Router::route(exchanger_ptr->_type, routing_key, binding.second->_binding_key))
                {
                    // 4. 消息发布（消息的管理）
                    DBG_LOG("route message to queue: %s\n", binding.first.c_str());
                    _virtual_host_ptr->publish(binding.first, properties_ptr, req->body());

                    // 5. 向线程池中添加一个消息消费任务（向订阅了当前消息队列的消费者推送一条消息，由线程池来完成
                    auto task = std::bind(&Channel::consumeTask, this, binding.first);
                    _thread_pool_ptr->push(task);       
                }
            }
            responseHandler(true, req->request_id(), req->channel_id());
        }

        // 消息的确认
        void ack(const AckRequestPtr &req)
        {
            _virtual_host_ptr->ack(req->queue_name(), req->message_id());
            responseHandler(true, req->request_id(), req->channel_id());
        }

        // 订阅队列消息
        void consume(const ConsumeRequestPtr &req)
        {
            // 1. 判断消息队列是否存在
            if(!_virtual_host_ptr->hasMsgQueue(req->queue_name()))
                responseHandler(false, req->request_id(), req->channel_id());

            // 2. 创建消费者进行管理
            Consumer::Callback cb = std::bind(&Channel::callback, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
            _consumer_manager_ptr->increaseConsumer(req->consumer_tag(), req->queue_name(), req->auto_ack(), cb);
            responseHandler(true, req->request_id(), req->channel_id());
        }

        // 取消订阅
        void cancel(const CancelRequestPtr& req)
        {
            _consumer_manager_ptr->removeConsumer(req->queue_name(), req->consumer_tag());
            responseHandler(true, req->request_id(), req->channel_id());
        }

    private:
        // 回调函数：消费者处理消息的方法
        void callback(const std::string& tag, const BasicProperties *bp, const std::string& body)
        {
            // 构建推送消息请求，将消息推送给对应的channel客户端
            ConsumeResponse rsp;
            rsp.set_channel_id(_id);
            rsp.set_consumer_tag(tag);
            rsp.set_body(body);
            if(bp) // 如果消息的属性存在，则给响应中设置
            {
                rsp.mutable_properties()->set_id(bp->id());
                rsp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                rsp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec_ptr->send(_conn_ptr, rsp);
            DBG_LOG("message push done\n");
        }

        // 消费消息任务
        void consumeTask(const std::string& queue_name)
        {
            // 1. 从队列中取出一条消息
            MessagePtr msg_ptr = _virtual_host_ptr->consume(queue_name);
            DBG_LOG("consume message from %s\n", queue_name.c_str());
            if(msg_ptr == nullptr)
            {
                DBG_LOG("failed to execute message task, the queue %s has no message\n", queue_name.c_str());
                return;
            }

            // 2. 从消费者队列中选择一个消费者
            Consumer::ConsumerPtr consumer_ptr = _consumer_manager_ptr->choose(queue_name);
            if(consumer_ptr == nullptr)
            {
                DBG_LOG("the queue %s has no consumer\n", queue_name.c_str());
                return;
            }

            // 3. 执行消费者处理消息的方式
            consumer_ptr->_callback(consumer_ptr->_tag, msg_ptr->mutable_payload()->mutable_properties(), msg_ptr->mutable_payload()->body());


            // 4. 判断消费者是否为自动确认，如果是自动确认，直接删除消息，否则需要服务器等待消费者的确认请求之后再删除
            if(consumer_ptr->_auto_ack)
                _virtual_host_ptr->ack(queue_name, msg_ptr->payload().properties().id());
        }

        // 响应处理函数 
        void responseHandler(bool success, const std::string& response_id, const std::string& channel_id)
        {
            // 构建响应并从指定channel发送
            CommonResponse response;
            response.set_success(success);
            response.set_response_id(response_id);
            response.set_channel_id(channel_id);
            _codec_ptr->send(_conn_ptr, response);
        }

    private:
        std::string _id;                                           // 信道标识
        Consumer::ConsumerPtr _consumer_ptr;                       // 消费者
        ConsumerManager::ConsumerManagerPtr _consumer_manager_ptr; // 消费者管理句柄
        muduo::net::TcpConnectionPtr _conn_ptr;                    // 连接句柄
        ProtobufCodecPtr _codec_ptr;                                 // 序列化协议句柄
        VirtualHost::VirtualHostPtr _virtual_host_ptr;             // 虚拟机句柄
        ThreadPool::ThreadPoolPtr _thread_pool_ptr;                // 线程池句柄
    };


    // 信道管理类
    class ChannelManager
    {
    public:
        using ChannelManagerPtr = std::shared_ptr<ChannelManager>;

    public:
        // 打开信道
        bool openChannel(const std::string& id, const ConsumerManager::ConsumerManagerPtr& consumer_manager_ptr, 
            const muduo::net::TcpConnectionPtr& conn_ptr, const ProtobufCodecPtr& codec_ptr, 
            const VirtualHost::VirtualHostPtr& virtual_host_ptr, const ThreadPool::ThreadPoolPtr& thread_pool_ptr)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if(_channels.count(id))
            {
                DBG_LOG("channel is exist, id is %s\n", id.c_str());
                return false;
            }
            Channel::ChannelPtr channel_ptr = std::make_shared<Channel>(id, consumer_manager_ptr, codec_ptr, conn_ptr, virtual_host_ptr, thread_pool_ptr);
            _channels[id] = channel_ptr;
            return true;
        }

        // 关闭信道
        void closeChannel(const std::string id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(id);
            if(it == _channels.end())
                DBG_LOG("channel is not exist, id is %s\n", id.c_str());
            else
                _channels.erase(it);
        }

        // 获取指定信道
        Channel::ChannelPtr getChannel(const std::string id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(id);
            if(it == _channels.end())
                return {};
            else
                return it->second;
        }
        
    private:
        std::unordered_map<std::string, Channel::ChannelPtr> _channels; // 信道映射表<信道标签，信道对象>
        std::mutex _mutex;
    };
}